package com.consumer;
import com.ai.util.ThreadLocalDateUtil;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.io.IOException;
import java.sql.Date;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @性能测试topicA producter
 * @version
 */

public class ConsumerPerfermanceAFile {
	private static ConsumerConnector consumer;
	public final static String TOPIC = "TestTopticA";
	static String sql = null;

	private void createConsumerConfig() {
		Properties props = new Properties();
		props.put("zookeeper.connect", "10.19.9.25:2181,10.19.9.26:2181,10.19.9.27:2181");
		props.put("zookeeper.session.timeout.ms", "5000");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.commit.interval.ms", "1000");
		// props.put("auto.offset.reset", "smallest");
		props.put("fetch.message.max.bytes", "2097152");
		props.put("group.id", "id1");
		ConsumerConfig config = new ConsumerConfig(props);
		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
	}

	public static void main(String[] args) throws SQLException, ParseException,
			IOException {
		System.out.println("receive message begin:");
		ConsumerPerfermanceAFile consumerThread = new ConsumerPerfermanceAFile();
		consumerThread.createConsumerConfig();
		consumerThread.consume(args[0], args[1]);//10  2
	}

	public void consume(String threadNum1, String threadNum2)
			throws IOException, ParseException {
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(TOPIC, new Integer(threadNum1));// 起20个线程
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
				.createMessageStreams(topicCountMap);
		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(TOPIC);
		System.out.println( ThreadLocalDateUtil.formatDate(new Date(
				System.currentTimeMillis())));
		ExecutorService executor = Executors.newFixedThreadPool(new Integer(threadNum2));
		for (int i = 1; i <= streams.size(); i++) {
			System.out.println("begin recvive" + i);
			executor.submit(new writeFileThread(streams.get(i - 1)));
		}



	}

}



